#include "RelOp.h"

void DuplicateRemoval::Run(Pipe &inPipe, Pipe &outPipe, Schema &mySchema){

    in = &inPipe;
    out = &outPipe;
    schema = &mySchema;

    pthread_create(&thread, NULL, DuplicateRemovalWorker, (void*)this);
}

void DuplicateRemoval::WaitUntilDone(){
    pthread_join(thread, NULL);
}

void DuplicateRemoval::Use_n_Pages(int n){
    nOfPages = n;
}

void DuplicateRemoval::DoWork(){
    OrderMaker om(schema);
    ComparisonEngine e;

     if (nOfPages == 0)
            nOfPages = 15;

    Pipe outS(100);
    BigQ q(*in, outS, om, nOfPages);

    Record current;
    Record *previous = new Record();
    outS.Remove(previous);

    while (outS.Remove(&current)) {
        if(e.Compare(previous, &current, &om) != 0){
          out->Insert(previous);
          previous->Copy(&current);
        }
    }

    out->Insert(previous);
    out->ShutDown();
}

void *DuplicateRemovalWorker(void *param){
     DuplicateRemoval duplicateRemoval = *((DuplicateRemoval*) param);
     duplicateRemoval.DoWork();
}
